import java.util.Map;
import java.util.HashMap;
import java.util.List;
import java.util.ArrayList;
import java.util.Base64;

import io.github.cdimascio.dotenv.Dotenv;
import lib.avro.Avro;
import lib.common.Common;
import org.apache.avro.Schema;

import proto.meta.MetaGrpc;
import proto.meta.MetaOuterClass.SchemaRead;
import proto.meta.MetaOuterClass.AvroGate;
import proto.xdefault.Xdefault.DefaultReply;

import java.io.*;
import io.grpc.stub.StreamObserver;

import com.linecorp.armeria.server.Server;
import com.linecorp.armeria.common.HttpResponse;
import com.linecorp.armeria.server.grpc.GrpcService;
import com.linecorp.armeria.common.HttpStatus;

// import com.linecorp.armeria.common.grpc.GrpcSerializationFormats; 
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

public class App {
  private static final String MODULE = "meta-server-java";
  private static final Logger LOGGER = LogManager.getLogger(LogManager.ROOT_LOGGER_NAME);

  private static void info(String message) {
    LOGGER.info("[component:{}] {}", MODULE, message);
  }

  private static void warn(String message) {
    LOGGER.warn("[component:{}] {}", MODULE, message);
  }

  private static void error(String message) {
    LOGGER.error("[component:{}] {}", MODULE, message);
  }

  static class MetaService extends MetaGrpc.MetaImplBase {
    @Override
    public void readRawSchema(SchemaRead read, StreamObserver<DefaultReply> ob) {
      int code = 0;
      String origin = "";
      String message = "";
      String details = "";
      String schemaFull = read.getSchema();
      String schema = schemaFull.split(":")[0];

      try {
        String selector = "";
        if (schemaFull.split(":", 2).length == 2) {
          selector = schemaFull.split(":", 2)[1];
        }
        String schemaType = schema.split("\\.")[0];
        if (!schema.contains("."))
          schema += ".";
        String schemaPath = String.format("meta-lib-static/%s.%s:%s", schema.replace(".", "/"), schemaType, selector);
        message = Avro.readRawSchema(schemaPath);

      } catch (Exception e) {
        e.printStackTrace();
        code = 3;
        message = "schema read error:" + e;
      }

      DefaultReply reply = DefaultReply.newBuilder()
          .setCode(code)
          .setMessage(message)
          .setDetails(details)
          .setOrigin(origin)
          .build();
      ob.onNext(reply);
      ob.onCompleted();
    }

    public Object cleanMeta(Object json) {
      if (json instanceof Map) {
        Map<String, Object> newJson = new HashMap<String, Object>();
        for (Map.Entry<String, Object> e : ((Map<String, Object>) json).entrySet()) {
          if (!e.getKey().startsWith("_")) {
            newJson.put(e.getKey(), cleanMeta(e.getValue()));
          }
        }
        return newJson;
      } else if (json instanceof List) {
        List<Object> newJson = new ArrayList<Object>();
        for (Object item : (List<Object>) json) {
          newJson.add(cleanMeta(item));
        }
        return newJson;
      } else {
        return json;
      }
    }

    @Override
    public void readAvscSchema(SchemaRead read, StreamObserver<DefaultReply> ob) {
      int code = 0;
      String origin = "";
      String message = "";
      String details = "";
      String schemaFull = read.getSchema();
      String schema = schemaFull.split(":")[0];

      try {
        String selector = "";
        if (schemaFull.split(":", 2).length == 2) {
          selector = schemaFull.split(":", 2)[1];
        }
        String schemaType = schema.split("\\.")[0];
        if (!schema.contains("."))
          schema += ".";
        String schemaPath = String.format("meta-lib-static/%s.%s:%s", schema.replace(".", "/"), schemaType, selector);
        String schemaText = Avro.readAvscSchema(schemaPath).toString();

        message = new ObjectMapper()
            .writeValueAsString(cleanMeta(new ObjectMapper().readValue(schemaText, Object.class)));
      } catch (Exception e) {
        e.printStackTrace();
        code = 3;
        message = "schema read error:" + e;
      }

      DefaultReply reply = DefaultReply.newBuilder()
          .setCode(code)
          .setMessage(message)
          .setDetails(details)
          .setOrigin(origin)
          .build();
      ob.onNext(reply);
      ob.onCompleted();
    }

    @Override
    public void gateJsonAvro(AvroGate gate, StreamObserver<DefaultReply> ob) {
      int code = 0;
      String message = "";
      String details = "";
      String origin = "";

      String schemaFull = gate.getSchema();
      String schema = schemaFull.split(":")[0];
      String json = gate.getBody();

      PrintStream stdout = System.out;
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      System.setOut(new PrintStream(baos));

      try {
        String selector = "";
        if (schemaFull.split(":", 2).length == 2) {
          selector = schemaFull.split(":", 2)[1];
        }
        String schemaType = schema.split("\\.")[0];
        if (!schema.contains("."))
          schema += ".";
        String schemaPath = String.format("meta-lib-static/%s.%s:%s", schema.replace(".", "/"), schemaType, selector);
        Schema bulkSchema = Avro.readAvscSchema(schemaPath);
        String jsonPatch = Avro.patchIos(json);

        Avro.fromJsonDebug(bulkSchema, jsonPatch);

      } catch (Exception e) {
        e.printStackTrace();
        code = 3;
        message = e.getMessage();
        details = baos.toString();
      } finally {
        System.setOut(stdout);
      }

      DefaultReply reply = DefaultReply.newBuilder()
          .setCode(code)
          .setMessage(message)
          .setDetails(details)
          .setOrigin(origin)
          .build();

      ob.onNext(reply);
      ob.onCompleted();
    }

    @Override
    public void gateBinAvro(AvroGate gate, StreamObserver<DefaultReply> ob) {
      int code = 0;
      String message = "";
      String details = "";
      String origin = "";

      String schemaFull = gate.getSchema();
      String schema = schemaFull.split(":")[0];
      String avroB64 = gate.getBody();

      PrintStream stdout = System.out;
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      System.setOut(new PrintStream(baos));
      try {
        String selector = "";
        if (schemaFull.split(":", 2).length == 2) {
          selector = schemaFull.split(":", 2)[1];
        }
        String schemaType = schema.split("\\.")[0];
        if (!schema.contains("."))
          schema += ".";
        String schemaPath = String.format("meta-lib-static/%s.%s:%s", schema.replace(".", "/"), schemaType, selector);
        Schema bulkSchema = Avro.readAvscSchema(schemaPath);

        byte[] avro_zstd = Base64.getDecoder().decode(avroB64.getBytes());
        Avro.fromCompressedBin(bulkSchema, avro_zstd);
      } catch (Exception e) {
        e.printStackTrace();
        code = 3;
        message = e.getMessage();
        details = baos.toString();
      } finally {
        System.setOut(stdout);
      }

      DefaultReply reply = DefaultReply.newBuilder()
          .setCode(code)
          .setMessage(message)
          .setDetails(details)
          .setOrigin(origin)
          .build();
      ob.onNext(reply);
      ob.onCompleted();
    }
  }

  public static void main(String[] args) throws Exception {
    Dotenv dotenv = Dotenv.configure().directory("meta-lib-static/env/.env").load();
    String hosts = Common.resolveEnv(dotenv, "META_SERVER_HOSTS", "127.0.0.1");
    String myIp = Common.resolveMyIp(hosts);
    if (myIp == null) {
      error("META_SERVER_HOSTS not match! skip");
      System.exit(0);
    }

    String portStr = Common.resolveEnv(dotenv, "META_SERVER_PORT", null);
    if (portStr == null) {
      error("META_SERVER_PORT not exist!");
      System.exit(1);
    }
    int port = Integer.parseInt(portStr);
    info(String.format("start server at %s", port));

    Server.builder()
        .http(port)
        .service("/meta/health", (ctx, req) -> HttpResponse.of(HttpStatus.OK))
        .service(
            GrpcService.builder()
                .addService(new MetaService())
                .build())
        .build().start();
  }
}
